package nsq

import (
	"github.com/nsqio/go-nsq"
	"log/lg"
	"time"
)

//初始化全局消息队列
var producer *nsq.Producer
var err error

//初始化接收tail日志的通道
var logDataChan = make(chan *logDate, 10000)

//初始化监控数据采集的通道
var logGoPs = make(chan []byte, 10000)

//接收日志的nsq通道的结构体
type logDate struct {
	topic string
	data  string
}

//接收数据采集的nsq通道结构体

// Init  生产初始化nsq
func Init(addr string) {

	producer, err = nsq.NewProducer(addr, nsq.NewConfig())
	if err != nil {
		lg.Log.Error("请检查nsq配置", err)
		return
	}
	err = producer.Ping()
	if err != nil {
		lg.Log.Panic("请检查nsq配置", err)
	}
	lg.Log.Info("初始化nsq成功!!!")

	//在后台启动此函数,等待通道的数据进行执行
	//开启三个并发携程进行执行
	for i := 1; i < 4; i++ {
		go sendNsq(i)
	}

	lg.Log.Info("执行后台队列nsq成功!!!")
	return
}

//从通道中拿取数据,往nsq发送数据
func sendNsq(i int) {

	for {
		select {
		case msg := <-logGoPs:
			//消费通道中的数据往nsq 发送
			err = producer.Publish("gops", msg)
			if err != nil {
				lg.Log.Error("nsq发送失败:", err)
				return
			}
			lg.Log.Info("nsq数据发送成功:", string(msg), "，当前执行线程：", i)

		case msg := <-logDataChan:
			//消费通道中的数据往nsq 发送
			err = producer.Publish(msg.topic, []byte(msg.data))
			if err != nil {
				lg.Log.Error("nsq发送失败:", err)
				return
			}
			lg.Log.Info("nsq数据发送成功:", msg, "，当前执行线程：", i)

		default:
			time.Sleep(time.Millisecond * 500)
			//	lg.Log.Error("没有获取到logDataChan中的通道数据")
		}
	}
}

// SendChan 给外部暴露的函数,此函数只把tail日志数据发送到通道中
func SendChan(data, topic string) {

	msg := &logDate{
		topic: topic,
		data:  data,
	}

	logDataChan <- msg

}

// SendGoPs 给外部暴露的函数,此函数只把监控日志数据发送到通道中
func SendGoPs(gops []byte) {
	logGoPs <- gops
}
